package com.boarsoft.boar.batch.demo;

import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.boarsoft.boar.batch.demo.biz.Coredpfm01Biz;
import com.boarsoft.boar.batch.demo.entity.Coredpfm01;
import com.boarsoft.boar.batch.demo.util.ToBigDecimal;
import com.boarsoft.common.util.RandomUtil;
import com.boarsoft.common.util.StreamUtil;
import com.boarsoft.flow.core.bean.ProcessHandler;

public class BigLoader01 implements ProcessHandler {
	private static final Logger log = LoggerFactory.getLogger(BigLoader01.class);

	private static final int BATCH_SIZE = 2000;

	@Autowired
	protected Coredpfm01Biz coredpfm01Biz;

	protected String fileName;

	protected ExecutorService threadPool = new ThreadPoolExecutor(//
			80, 80, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(80),
			Executors.defaultThreadFactory(), new ThreadPoolExecutor.CallerRunsPolicy());

	class SubSaver implements Runnable {
		int index;
		List<Coredpfm01> list;

		public SubSaver(int i, List<Coredpfm01> a) {
			index = i;
			list = a;
		}

		@Override
		public void run() {
			try {
				coredpfm01Biz.batchSave(index, list);
			} catch (SQLException e) {
				log.error("Error on batch save", e);
			}
		}
	}

	@Override
	public Object process(String entry, Object data, Throwable throwable) throws Throwable {
		Map<Integer, List<Coredpfm01>> map = new HashMap<Integer, List<Coredpfm01>>();
		for (int i = 0; i < 10; i++) {
			map.put(i, new ArrayList<Coredpfm01>());
		}
		File file = new File(fileName);
		BufferedInputStream bis = null;
		BufferedReader br = null;
		long l = System.currentTimeMillis();
		try {
			bis = new BufferedInputStream(new FileInputStream(file));
			br = new BufferedReader(new InputStreamReader(bis, "utf-8"), 8 * 1024 * 1024);
			String ln = null;
			while ((ln = br.readLine()) != null) {
				Coredpfm01 o = new Coredpfm01();
				try {
					this.parse(ln, o);
				} catch (Exception e) {
					log.error("Error on parse string:\n{}\n", ln, e);
					continue;
				}
				int i = Math.abs(o.getId().hashCode()) % 8;
				List<Coredpfm01> lt = map.get(i);
				lt.add(o);
				if (lt.size() >= BATCH_SIZE) {
					threadPool.execute(new SubSaver(i, lt));
					map.put(i, new ArrayList<Coredpfm01>());
				}
			}
			for (int i : map.keySet()) {
				List<Coredpfm01> lt = map.get(i);
				if (lt.isEmpty()) {
					continue;
				}
				threadPool.execute(new SubSaver(i, lt));
			}
			log.info("...............completed in {} minutes..............", //
					(System.currentTimeMillis() - l) / 1000 / 60);
		} catch (IOException e) {
			log.error("Error on read line from {}", fileName, e);
		} finally {
			// StreamUtil.close(bis);
			StreamUtil.close(br);
		}
		return true;
	}

	private void parse(String s, Coredpfm01 o) {
		String[] a = s.split("\03");
		o.setId(RandomUtil.randomUUID());
		o.setDM01ACCT(a[0]);
		o.setDM01CLSFLG(a[1]);
		o.setDM01DOCLST(a[2]);
		o.setDM01PSWLST(a[3]);
		o.setDM01ACTFLG(a[4]);
		o.setDM01SECFLG(a[5]);
		o.setDM01DOCFLG(a[6]);
		o.setDM01VOUCLS(a[7]);
		o.setDM01VOUNUM(a[8]);
		o.setDM01WTTYPE(a[9]);
		o.setDM01WTPRF1(a[10]);
		o.setDM01INQPAS(a[11]);
		o.setDM01WTPRF3(a[12]);
		o.setDM01PRTNUM(ToBigDecimal.stringToBigDecimal(a[13]));
		o.setDM01UPRTNU(ToBigDecimal.stringToBigDecimal(a[14]));
		o.setDM01ACCNUM(ToBigDecimal.stringToBigDecimal(a[15]));
		o.setDM01ACCSEQ(ToBigDecimal.stringToBigDecimal(a[16]));
		o.setDM01ORGNO(a[17]);
		o.setDM01CLIPRO(a[18]);
		o.setDM01PBNUM(ToBigDecimal.stringToBigDecimal(a[19]));
		o.setDM01LICNO(a[20]);
		o.setDM01AUTNO(a[21]);
		o.setDM01AVLDAT(a[22]);
		o.setDM01FLAG(a[23]);
		o.setDM01NODE(a[24]);
		o.setLAST_ETL_ACG_DT(a[25]);
		o.setDEL_F(ToBigDecimal.stringToLong(a[26]));
	}

	public String getFileName() {
		return fileName;
	}

	public void setFileName(String fileName) {
		this.fileName = fileName;
	}
}
